package v1

import (
	"chat/app/common"
	"chat/app/dao"
	"chat/setting"
	"chat/utils/logger"
	"context"
	"encoding/json"
	"fmt"
	mapset "github.com/deckarep/golang-set/v2"
	"github.com/gorilla/websocket"
	"github.com/petermattis/goid"
	"log"
	"net"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	CmdSingleMsg  = 10                    // 单聊
	CmdRoomMsg    = 11                    // 群聊
	CmdHeart      = 0                     //心跳
	ServerIpName  = "chat_user_"          // 在线注册用户前缀
	ULeaveMessage = "chat_leave_message_" // 留言前缀
	USendMessage  = "chat_send_message_"  // 消息发送前缀
)

var (
	// 当前服务器ip+port
	serverIp string

	// 协议参数
	upGrader = websocket.Upgrader{
		ReadBufferSize:  1024, // 读的缓冲大小
		WriteBufferSize: 1024, // 写的缓冲大小
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}
)

// Data 消息数据结构
// "`omitempty`"是Go语言中的一个标记，用于在编码和解码JSON数据时控制字段的行为。它可以应用于结构体字段的标签中。
// 当字段被标记为"`omitempty`"时，它表示在编码JSON数据时，如果该字段的值为空值，即零值或空引用，则忽略该字段，不将其包含在生成的JSON中。
// 换句话说，如果字段的值为零值或空引用，则不会在JSON中创建对应的键值对。
// 这个特性对于减少生成的JSON大小、提高网络传输效率以及使JSON数据更加简洁和可读是有用的。
type Data struct {
	Ip         string `json:"ip,omitempty" form:"ip"`
	Type       string `json:"type,omitempty" form:"type"`
	Uid        int64  `json:"uid,omitempty" form:"uid"`                 // 消息发送者
	ReceiveUid int64  `json:"receive_uid,omitempty" form:"receive_uid"` // 消息接收者 单聊和群聊转化处理后所对应的接收人uid
	Nickname   string `json:"nickname,omitempty" form:"nickname"`
	Avatar     string `json:"avatar,omitempty" form:"avatar"`
	Sex        int8   `json:"sex,omitempty" form:"sex"`
	Remark     string `json:"remark,omitempty" form:"remark"`
	Message
}
type Message struct {
	Id      int64  `json:"id,omitempty" form:"id"`           //消息ID
	Userid  int64  `json:"userid,omitempty" form:"userid"`   //谁发的
	Cmd     int    `json:"cmd,omitempty" form:"cmd"`         //群聊还是私聊
	Dstid   int64  `json:"dstid,omitempty" form:"dstid"`     //对端用户ID/群ID
	Media   int    `json:"media,omitempty" form:"media"`     //消息按照什么样式展示
	Content string `json:"content,omitempty" form:"content"` //消息的内容
	Pic     string `json:"pic,omitempty" form:"pic"`         //预览图片
	Url     string `json:"url,omitempty" form:"url"`         //服务的URL
	Memo    string `json:"memo,omitempty" form:"memo"`       //简单描述
	Amount  int    `json:"amount,omitempty" form:"amount"`   //其他和数字相关的
}

// connection 连接
type connection struct {
	mut       sync.Mutex        // output cloe 锁
	mu        sync.Mutex        // 消息 write 锁
	ws        *websocket.Conn   // 连接conn
	output    chan []byte       // 串行输出消息
	groupSets mapset.Set[int64] // 群组数组
	data      *Data             // 消息格式
	uid       int64             // 当前连接uid
	userInfo  map[string]any    // userInfo
	closed    bool              // close output
	logs      *logger.Logger    // *logger.Logger
	requestId any               // requestId
}

// 共享数据
type hub struct {
	// connection map 锁
	mu sync.RWMutex
	// connection 连接
	connections map[int64]*connection
	// 从服务器发送的信息
	broadcast chan *Data
	//broadcast chan []byte
	//broadcast chan *connection
	// 从连接器注册请求
	register chan *connection
	// 销毁请求
	unregister chan *connection
}

// 初始化
var h = &hub{
	connections: make(map[int64]*connection),
	broadcast:   make(chan *Data),
	//broadcast:   make(chan []byte),
	//broadcast:  make(chan *connection),
	register:   make(chan *connection),
	unregister: make(chan *connection),
}

func init() {
	// 服务器ip+port
	serverIp = getHostIp() + ":" + setting.Server.Port
	// 分解消息
	go h.Run()
}

func getHostIp() string {
	conn, err := net.Dial("udp", "8.8.8.8:53")
	if err != nil {
		fmt.Println("get current host ip err: ", err)
		return ""
	}
	addr := conn.LocalAddr().(*net.UDPAddr)
	ip := strings.Split(addr.String(), ":")[0]
	return ip
}

// SendMessage 发送消息
func SendMessage() {
	for {
		key := USendMessage + serverIp
		message, err := dao.BLPop(key, 100*time.Second)
		log.Printf("sendMessage message: %v, err: %v \n", message, err)
		if err != nil {
			//break
			if err.Error() != "NOAUTH Authentication required." {
				time.Sleep(1 * time.Second)
				continue
			}
		}
		if message != "" {
			data := Data{}
			json.Unmarshal([]byte(message), &data)
			receiveUid := data.ReceiveUid // 消息接收人
			h.sendMsg(receiveUid, []byte(message))
			log.Printf("sendMessage message send: %v \n", message)
		}
	}
}

// 发送到消息管道
func (h *hub) sendMsg(receiveUid int64, msg []byte) (ok bool) {
	h.mu.Lock()
	c, ok := h.connections[receiveUid]
	h.mu.Unlock()
	logger.Info("websocket-chain sendMsg 666666 receiveUid, %v, ok: %v", receiveUid, ok)
	if ok {
		// 在线,直接发送消息
		select {
		case c.output <- msg:
		default:
			logger.Info("websocket-chain sendMsg close receiveUid: %v", receiveUid)
			h.mu.Lock()
			delete(h.connections, receiveUid)
			h.mu.Unlock()
			// 清除redis关联的uid
			uidStr := strconv.FormatInt(receiveUid, 10)
			dao.Del(ServerIpName + uidStr)
			c.mut.Lock()
			if !c.closed {
				close(c.output)
				c.closed = true
			}
			c.mut.Unlock()
		}
	}
	return
}

func heartbeat(wsc *connection, uid int64) {
	msg := []byte("heartbeat")
	uidStr := strconv.FormatInt(uid, 10)
	for {
		//leaveMessageSend(wsc.uid)
		time.Sleep(90 * time.Second)
		wsc.mu.Lock()
		err := wsc.ws.WriteMessage(websocket.TextMessage, msg)
		wsc.mu.Unlock()
		if err != nil {
			wsc.logs.Info("websocket-chain heartbeat close, uid: %v", uid)
			return
		}
		//dao.Expire(ServerIpName+uidStr, 120*time.Second)
		dao.SetEX(ServerIpName+uidStr, serverIp, 120*time.Second)
		//log.Printf("heartbeat")
		wsc.logs.Info("heartbeat, err: %v", err)
	}
}

// PostMessage时的消息顺序问题???
// h.register 是否能满足需求???
func leaveMessageSend(wsc *connection) {
	userIdStr := strconv.FormatInt(wsc.uid, 10)
	msgCount, _ := dao.LLen(ULeaveMessage + userIdStr)
	log.Printf("leaveMessageSend msgCount: %v \n", msgCount)
	if msgCount > 0 {
		for i := 0; i <= int(msgCount); i++ {
			message, _ := dao.LPop(ULeaveMessage + userIdStr)
			if message != "" {
				//h.sendMsg(uid, []byte(message))
				wsc.output <- []byte(message)
				log.Printf("leaveMessageSend message: %v \n", message)
			}
		}
	}
}

func (h *hub) Run() {
	for {
		select {
		case wsc := <-h.register: // 注册在线用户
			// 获取该用户全部群Id todo 可以删除了???
			//contacts, rowsAffected := dao.SearchContactIds(wsc.uid)
			//if rowsAffected {
			//	for _, v := range *contacts {
			//		wsc.groupSets.Add(v.ReceiveId)
			//	}
			//}
			// 注册在线用户
			h.mu.Lock()
			h.connections[wsc.uid] = wsc

			// 存储在线用户和服务器ip到redis,实现分布式集群时消息的推送
			uidStr := strconv.FormatInt(wsc.uid, 10)
			dao.SetString(ServerIpName+uidStr, serverIp, 120*time.Second)
			//ipStr, _ := dao.GetString(ServerIpName + uidStr)
			//fmt.Println("ipStr：" + ipStr.(string))

			// 组装data数据 发送欢迎信息
			wsc.data.Ip = wsc.ws.RemoteAddr().String()
			wsc.data.Type = "handshake"
			wsc.data.Uid = wsc.uid
			wsc.data.Content = string("welcome")
			dataByte, _ := json.Marshal(wsc.data)
			wsc.output <- dataByte
			// 查看是否有留言
			leaveMessageSend(wsc)
			h.mu.Unlock()
			// 心跳
			go heartbeat(wsc, wsc.uid)

		case wsc := <-h.unregister: // 注销在线用户
			// 判断map里是否存在要删的数据
			if _, ok := h.connections[wsc.uid]; ok {
				wsc.logs.Info("websocket-chain h.unregister close uid: %v", wsc.uid)
				h.mu.Lock()
				delete(h.connections, wsc.uid)
				h.mu.Unlock()
				// 清除redis关联的uid
				uidStr := strconv.FormatInt(wsc.uid, 10)
				dao.Del(ServerIpName + uidStr)
				// 关闭连接管道
				wsc.mut.Lock()
				if !wsc.closed {
					close(wsc.output)
					wsc.closed = true
				}
				wsc.mut.Unlock()
			}
		case data := <-h.broadcast: // 广播消息
			h.Connections(data)
		}
	}
}

// Connections 选择对的人广播消息
func (h *hub) Connections(data *Data) {
	uid := data.Uid // 当前发送者
	c, ok := h.connections[uid]
	logger.Info("websocket-chain 222222 uid: %v, ok: %v", uid, ok)
	//logger.Info("websocket-chain wsc: %v, ok: %v", c, ok)

	// goid
	goId := goid.Get()
	_, ok2 := common.RequestIdMap.Load(goId)
	if !ok2 {
		//common.RequestIdMap.Store(goId, c.requestId)
		logger.Ctx = context.WithValue(logger.Ctx, goId, c.requestId)
	}
	logger.Info("websocket-chain Connections goId: %v, wsc.requestId: %v", goId, c.requestId)

	if ok {
		log.Printf("Connections======3")
		c.dispatch(data)
		// 清理消息数据
		c.reset()
	} else {
		// 没找到自己
	}
}

// 广播所有人
//func (h *hub) broadcastsMessage(data []byte) {
//	for uid, c := range h.connections {
//		select {
//		case c.output <- data:
//		default:
//			// 防止死循环
//			mu := new(sync.Mutex)
//			mu.Lock()
//			delete(h.connections, uid)
//			mu.Unlock()
//			// 清除redis关联的uid
//			uidStr := strconv.FormatInt(uid, 10)
//			dao.Del(ServerIpName+uidStr)
//			close(c.output)
//		}
//	}
//}
//
//// 群组分发消息 userId当前登录uid
//func (h *hub) groupMsg(groupId int64, userId int64, msg []byte) {
//	// 给群发消息 遍历所有在线用户
//	//for _, v := range h.connections {
//	//	log.Printf("v.uid: %#v, v.data.Userid: %#v, v.groupSets.Contains(msg.Dstid): %#v \n",
//	//		v.uid, v.data.Userid, v.groupSets.Contains(groupId))
//	//	if v.groupSets.Contains(groupId) && v.uid != v.data.Userid {
//	//		v.output <- msg
//	//	}
//	//}
//	// 只查找对应群组下的人 可添加redis缓存
//	contacts, rowsAffected := dao.SearchContactUids(groupId)
//	if rowsAffected {
//		for _, v := range *contacts {
//			if v.OwnerId != userId {
//				h.sendMsg(v.OwnerId, msg)
//			}
//		}
//	}
//}

// 消息分发到发送服务
func (wsc *connection) searchSend(receiveId string, data *Data) {
	// todo ====== 保存消息到消息队列 (每个发送队列用对应的服务ip作为键) ======
	userId, _ := strconv.ParseInt(receiveId, 10, 64)
	data.ReceiveUid = userId // 设置消息接收人
	messageData, _ := json.Marshal(data)
	h.mu.RLock()
	_, ok := h.connections[userId]
	h.mu.RUnlock()
	if ok {
		key := USendMessage + serverIp
		wsc.logs.Info("websocket-chain searchSend 333333 receiveId: %v, serverIp: %v", receiveId, serverIp)
		dao.RPush(key, messageData)
	} else {
		hostIp, _ := dao.GetString(ServerIpName + receiveId)
		wsc.logs.Info("websocket-chain searchSend 444444 receiveId: %v, hostIp: %v", receiveId, hostIp)
		if hostIp != "" {
			key := USendMessage + hostIp
			dao.RPush(key, messageData)
		} else {
			// 不在线,保存消息留言
			h.saveMessage(userId, messageData)
		}
	}

	// todo ====== 直接发送消息到客户端 ======
	//userId, _ := strconv.ParseInt(receiveId, 10, 64)
	//messageData, _ := json.Marshal(data)
	//ok := h.sendMsg(userId, messageData)
	////ok := false
	//wsc.logs.Info("websocket-chain searchSend 333333 receiveId: %v, messageData: %v", receiveId, string(messageData))
	//if !ok {
	//	serverIp, _ := dao.GetString(ServerIpName + receiveId)
	//	host := serverIp
	//	wsc.logs.Info("websocket-chain searchSend 444444 receiveId: %v, host: %v", receiveId, host)
	//	if host != "" {
	//		ip := getHostIp()
	//		ip = ip + ":" + setting.Server.Port
	//		wsc.logs.Info("websocket-chain searchSend 555555 receiveId: %v, host: %v, ip: %v", receiveId, host, ip)
	//		// 本地没有conn找相应服务器发送
	//		httpPost := func() {
	//			//token := wsc.content.Query("token") // 发送者token
	//			requestUrl := url.URL{
	//				Scheme:   "http",
	//				Host:     host,
	//				Path:     "/api/v1/postMessage",
	//				RawQuery: "token=",
	//			}
	//			//messageData, _ := json.Marshal(data)
	//			param := url.Values{}
	//			param.Set("receive_id", receiveId)
	//			param.Set("message", string(messageData))
	//			payload := strings.NewReader(param.Encode())
	//			resp, _ := http.Post(requestUrl.String(), "application/x-www-form-urlencoded", payload)
	//			log.Printf("respPost: %#v", resp)
	//			//if resp.Status == "200" {
	//			//	body, err := ioutil.ReadAll(resp.Body)
	//			//}
	//		}
	//		go httpPost()
	//	} else {
	//		// 不在线,保存消息留言
	//		h.saveMessage(userId, messageData)
	//	}
	//}
}

// 保存留言
func (h *hub) saveMessage(receiveId int64, msg []byte) {
	userId := strconv.FormatInt(receiveId, 10)
	dao.RPush(ULeaveMessage+userId, msg)
}

// 后端调度
func (wsc *connection) dispatch(data *Data) {

	switch data.Cmd {
	case CmdSingleMsg:
		// 给单人发消息
		wsc.logs.Info("websocket-chain 单聊...")
		dstid := strconv.FormatInt(data.Dstid, 10)
		wsc.searchSend(dstid, data)
	case CmdRoomMsg:
		// 只查找对应群组下的人 可添加redis缓存
		wsc.logs.Info("websocket-chain 群聊...")
		groupId := data.Dstid
		contacts, rowsAffected := dao.SearchContactUids(groupId)
		if rowsAffected {
			for _, v := range *contacts {
				if v.OwnerId != wsc.uid {
					dstid := strconv.FormatInt(v.OwnerId, 10)
					wsc.searchSend(dstid, data)
				}
			}
		}
	case CmdHeart:
		//检测客户端的心跳
		//log.Println("检测客户端的心跳...")
		wsc.logs.Info("websocket-chain 检查心跳...")

	default:
		// 广播所有好友
		wsc.logs.Info("websocket-chain 广播...")
		contacts, rowsAffected := dao.SearchFriendUids(wsc.uid)
		if rowsAffected {
			for _, v := range *contacts {
				dstid := strconv.FormatInt(v.ReceiveId, 10)
				wsc.searchSend(dstid, data)
			}
		}
	}

}

// Writer 发送消息
func (wsc *connection) Writer() {
	// 写入ws数据
	for message := range wsc.output {
		wsc.mu.Lock()
		err := wsc.ws.WriteMessage(websocket.TextMessage, message)
		wsc.mu.Unlock()
		if err != nil {
			continue
		}
	}
	wsc.ws.Close()
}

// Reader 读取消息
func (wsc *connection) Reader() {
	for {
		// 读取ws中的数据
		_, message, err := wsc.ws.ReadMessage()
		if err != nil {
			break
		}

		wsc.logs.Info("websocket-chain message: %v", string(message))
		//log.Printf("message: %v \n", string(message))

		wsc.reset()
		//log.Printf("wsc.data: %#v", *wsc.data)
		json.Unmarshal(message, &wsc.data)
		//log.Printf("message: %#v", wsc.data)

		switch wsc.data.Type {
		case "ping":
			wsc.data.Uid = wsc.uid
			wsc.data.Type = "pong"
			wsc.data.Cmd = 10
			wsc.data.Dstid = wsc.uid // 发送给自己
			h.broadcast <- wsc.data
		default:
			userInfo := wsc.userInfo
			wsc.data.Uid = wsc.uid
			wsc.data.Type = "message"
			//wsc.data.Ip = wsc.ws.RemoteAddr().String()
			wsc.data.Avatar = userInfo["avatar"].(string)
			wsc.data.Nickname = userInfo["nickname"].(string)
			wsc.data.Sex = userInfo["sex"].(int8)
			wsc.data.Remark = userInfo["remark"].(string)
			// 直接写入发送管道 这种方法适合单服务器方式
			messageData, _ := json.Marshal(wsc.data)
			wsc.logs.Info("websocket-chain 111111: %v", string(messageData))
			h.broadcast <- wsc.data
			// 第二种方法 集群方式 消息可直接写入redis队列 然后有一个专门的分发服务器 发送到对应的PostMessage 这样可保证消息顺序
			// 例如：dao.RPush(ULeaveMessage+userId, message)
		}
	}
}

// 重制数据初始值
func (wsc *connection) reset() {
	wsc.data.Type = ""
	wsc.data.Ip = ""
	wsc.data.Avatar = ""
	wsc.data.Nickname = ""
	wsc.data.Sex = 0
	wsc.data.Remark = ""
	wsc.data.Id = 0
	wsc.data.Userid = 0
	wsc.data.Cmd = 0
	wsc.data.Dstid = 0
	wsc.data.Media = 0
	wsc.data.Content = ""
	wsc.data.Pic = ""
	wsc.data.Url = ""
	wsc.data.Memo = ""
	wsc.data.Amount = 0
}

// PostMessage 收消息
func (c *App) PostMessage() {
	receive_id := c.PostForm("receive_id")
	receiveId, _ := strconv.ParseInt(receive_id, 10, 64)
	message := c.PostForm("message")
	ok := h.sendMsg(receiveId, []byte(message))
	if !ok {
		// 没有找到在线好友，保存消息留言
		h.saveMessage(receiveId, []byte(message))
	}
}

// Chat 消息主函数
func (c *App) Chat() {
	// 用户id
	uid := c.UserInfo["uid"]
	userId := uid.(int64)

	// traceId
	requestId, _ := c.Get("requestId")
	log.Printf("c.log.Keys[requestId]: %v, c.Keys[requestId]: %v", c.Logs.Keys["requestId"], c.Keys["requestId"])

	// 升级get请求为webSocket协议
	ws, err := upGrader.Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		return
	}

	// 设置连接等初始值
	wsc := &connection{
		ws:        ws,
		output:    make(chan []byte, 128),
		groupSets: mapset.NewSet[int64](),
		data:      new(Data),
		uid:       userId,
		userInfo:  c.Context.UserInfo,
		closed:    false,
		logs:      c.Logs,
		requestId: requestId,
	}
	c.Logs.Info("wsc: %v, c.Keys[requestId]: %v", wsc, c.Keys["requestId"])

	// 注册conn
	h.register <- wsc

	go wsc.Writer()
	wsc.Reader()

	// 连接断开后的清理操作
	defer func() {
		c.Logs.Info("websocket-chain chat defer: %v", wsc.uid)
		wsc.reset()
		wsc.data.Uid = wsc.uid
		wsc.data.Cmd = 0
		wsc.data.Type = "logout"
		wsc.data.Content = string("goodbye")
		h.broadcast <- wsc.data
		h.unregister <- wsc
		ws.Close()
	}()
}
